package rabbitmq

import (
	"app/util"
	"errors"
	"github.com/streadway/amqp"
	"net"
	"time"
)

type Channel struct {
	conn         *amqp.Connection
	channel      *amqp.Channel
	queue        amqp.Queue
	channelconf  *ChannelConf
	queueconf    *QueueConf
	exchangeconf *ExchangeConf
	pusherconf   *PusherConf
	poperconf    *PoperConf
}

func NewChannel(conf *ChannelConf) *Channel {

	config := amqp.Config{
		Heartbeat: time.Duration(10) * time.Second,
		Dial: func(network, addr string) (net.Conn, error) {
			c, err := net.DialTimeout(network, addr, time.Duration(3)*time.Second)
			if err != nil {
				return nil, err
			}
			return c, nil
		},
	}

	connection, err := amqp.DialConfig(conf.Connect, config)
	if err != nil {
		util.Error("NewRabbitMQ error")
		return nil
	}

	rq := &Channel{channelconf: conf, conn: connection}

	rq.channel, err = rq.conn.Channel()
	if err != nil {
		return nil
	}

	err = rq.channel.Qos(conf.QosCount, conf.QosSize, false)
	if err != nil {
		goto channelclose
	}

	return rq
channelclose:
	rq.channel.Close()
	return nil
}

func (rq *Channel) CloseChannel() {

	if rq.channel != nil {
		rq.channel.Close()
	}
	if rq.conn != nil {
		rq.conn.Close()
	}
}

func (rq *Channel) ACK(tag uint64, multiple bool) error {

	return rq.channel.Ack(tag, multiple)
}

func (rq *Channel) DeclareExchange(conf *ExchangeConf) (err error) {

	err = rq.channel.ExchangeDeclare(conf.Name, conf.Type, conf.Durable, conf.AutoDeleted, conf.Internal, conf.NoWait, nil)
	if err != nil {
		util.Warning("ExchangeDelete error")
		return errors.New("ExchangeDeclare error")
	}

	rq.exchangeconf = conf
	return nil
}

func (rq *Channel) DeleteExchange(conf *ExchangeConf) (err error) {

	err = rq.channel.ExchangeDelete(conf.Name, false, false)
	if err != nil {
		util.Warning("ExchangeDelete error")
		return errors.New("ExchangeDelete error")
	}
	rq.exchangeconf = nil
	return nil
}

func (rq *Channel) DeclareQueue(conf *QueueConf) (err error) {

	//处理x-message-ttl的类型，json里面写的是int，go读出来的是double
	if _, ok := conf.Args["x-message-ttl"]; ok {
		t := int32(conf.Args["x-message-ttl"].(float64))
		delete(conf.Args, "x-message-ttl")
		conf.Args["x-message-ttl"] = t
	}

	rq.queue, err = rq.channel.QueueDeclare(conf.Name, conf.Durable, conf.AutoDelete, conf.Exclusive, conf.NoWait, conf.Args)
	if err != nil {
		//return errors.New("QueueDeclare error")
		util.Warning("Queue IsExist")
		rq.queueconf = conf
		return nil
	}

	for _, b := range conf.Bind {
		if err = rq.channel.QueueBind(conf.Name, b.Key, b.ExchangeName, b.NoWait, nil); err != nil {
			return err
		}
	}

	rq.queueconf = conf
	return nil
}

func (rq *Channel) DeleteQueue(conf *QueueConf) (err error) {

	for _, b := range conf.Bind {
		if err = rq.channel.QueueUnbind(conf.Name, b.Key, b.ExchangeName, nil); err != nil {
			return err
		}
	}

	rq.queueconf = nil
	return nil
}

func (rq *Channel) DeclarePush(conf *PusherConf) {
	rq.pusherconf = conf
}

func (rq *Channel) DeletePush(conf *PusherConf) {
	rq.pusherconf = nil
}

func (rq *Channel) DeclarePop(conf *PoperConf) {
	rq.poperconf = conf
}

func (rq *Channel) DeletePop(conf *PoperConf) {
	rq.poperconf = nil
}

func (rq *Channel) GetExchangeConf() *ExchangeConf {
	return rq.exchangeconf
}

func (rq *Channel) GetQueueConff() *QueueConf {

	return rq.queueconf
}
